package org.neptune.service.service.impl;

import java.util.Date;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.neptune.api.entity.*;
import org.neptune.api.request.SeckillRequest;
import org.neptune.api.response.SeckillResponse;
import org.neptune.common.constant.CharacterConstant;
import org.neptune.common.constant.NeptuneConstant;
import org.neptune.common.enums.SeckillStatus;
import org.neptune.service.annotation.SeckillActionLimit;
import org.neptune.service.annotation.SeckillActionLock;
import org.neptune.service.config.NeptuneConfig;
import org.neptune.service.dao.SeckillMapper;
import org.neptune.service.rabbit.LogProducer;
import org.neptune.service.rabbit.SeckillDlxProducer;
import org.neptune.service.rabbit.SeckillProducer;
import org.neptune.service.redis.RedisRepository;
import org.neptune.service.redis.RedissonUtil;
import org.neptune.service.service.SeckillService;
import org.neptune.service.util.WebUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class SeckillServiceImpl implements SeckillService {

	private static final Logger logger = LoggerFactory.getLogger(SeckillServiceImpl.class);

	@Autowired
	private SeckillMapper seckillMapper;

	@Autowired
	private LogProducer logProducer;

	@Autowired
	private RedisRepository redisRepository;

	@Autowired
	private SeckillProducer seckillProducer;

	@Autowired
	private SeckillDlxProducer seckillDlxProducer;

	@Autowired
	private NeptuneConfig neptuneConfig;

	// 获取可用处理器数量，设置线程池大小
	private static int thread_count = Runtime.getRuntime().availableProcessors();

	private static ThreadPoolExecutor executor = new ThreadPoolExecutor(thread_count,
			thread_count + 1, 10L, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());

	private Lock lock = new ReentrantLock(true);

	@Override
	// 限流
	@SeckillActionLimit
	@Transactional
	public SeckillResponse seckillByAopLimit(SeckillRequest seckillRequest) {
		logger.info("current thread name : {}", Thread.currentThread().getName());
		SeckillResponse seckillResponse = new SeckillResponse();
		int count = this.querySeckillGoodsCount(seckillRequest);
		logger.info("current goods count : {}", count);
		if (count > 0){
			logger.info("userid : {}; goodsId : {}", seckillRequest.getUserId(),
					seckillRequest.getSeckillId());
			try{
				if (this.updateSeckillGoods(seckillRequest, count - 1)){
					seckillResponse.setMsg(SeckillStatus.SUCCESS.getMsg());
				} else{
					seckillResponse.setMsg(SeckillStatus.ERROR.getMsg());
				}
			} catch (Exception e){
				logger.error("操作失败：", e);
				seckillResponse.setMsg(SeckillStatus.ERROR.getMsg());
			}
			logger.info("#########start producer,userid : {}", seckillRequest.getUserId());
			logProducer.sender(this.generateSeckillLog(seckillRequest));
			logger.info("#########end producer,userid : {}", seckillRequest.getUserId());
		} else{
			seckillResponse.setMsg(SeckillStatus.END.getMsg());
		}
		return seckillResponse;
	}

	@Override
	public SeckillResponse seckillByReentranLock(SeckillRequest seckillRequest) {

		SeckillResponse seckillResponse = new SeckillResponse();
		try{
			// 锁上移，保证事务完整
			lock.lock();
			int count = this.querySeckillGoodsCount(seckillRequest);
			if (count > 0){

				if (this.updateSeckillGoodsByLock(seckillRequest, count)){
					seckillResponse.setMsg(SeckillStatus.SUCCESS.getMsg());
				} else{
					seckillResponse.setMsg(SeckillStatus.ERROR.getMsg());
				}
				logProducer.sender(this.generateSeckillLog(seckillRequest));
			} else{
				seckillResponse.setMsg(SeckillStatus.END.getMsg());
			}
		} catch (Exception e){
			logger.error("操作失败：", e);
			seckillResponse.setMsg(SeckillStatus.ERROR.getMsg());
		} finally{
			lock.unlock();
		}
		return seckillResponse;
	}

	/**
	 * aop的执行顺序，order
	 * <p>
	 * 如果事务先执行--？？
	 *
	 * @param seckillRequest
	 * @return
	 */
	@Override
	@SeckillActionLock
	@Transactional
	public SeckillResponse seckillByAopLock(SeckillRequest seckillRequest) {

		int count = this.querySeckillGoodsCount(seckillRequest);
		if (count > 0){
		}

		return null;
	}

	/**
	 * 查询秒杀商品库存
	 *
	 * @param seckillRequest
	 * @return
	 */
	private int querySeckillGoodsCount(SeckillRequest seckillRequest) {

		SeckillExample example = new SeckillExample();

		example.createCriteria().andSeckillIdEqualTo(seckillRequest.getSeckillId());

		List<Seckill> seckills = seckillMapper.selectByExample(example);

		if (seckills.isEmpty()){
			return 0;
		}

		return seckills.get(0).getNumber();
	}

	private int querySeckillGoodsCountForUpdate(SeckillRequest seckillRequest) {

		return seckillMapper.querySeckillGoodsCountForUpdate(seckillRequest.getSeckillId());
	}

	/**
	 * 执行秒杀
	 *
	 * @return
	 */
	@Override
	public boolean updateSeckillGoods(SeckillRequest seckillRequest, int count) {

		Seckill seckill = new Seckill();
		seckill.setNumber(count - 1);
		SeckillExample example = new SeckillExample();
		example.createCriteria().andSeckillIdEqualTo(seckillRequest.getSeckillId());

		logger.info("update seckill goods id :{} ", seckillRequest.getSeckillId());

		return seckillMapper.updateByExampleSelective(seckill, example) > 0;
	}

	/**
	 * 执行秒杀
	 * <p>
	 * 
	 * @Transactional 加于private方法, 无效
	 *                <p>
	 * @Transactional 加于未加入接口的public方法, 再通过普通接口方法调用, 无效
	 *                <p>
	 * @Transactional 加于接口方法, 无论下面调用的是private或public方法, 都有效
	 *                <p>
	 * @Transactional 加于接口方法后, 被本类普通接口方法直接调用, 无效
	 *                <p>
	 * @Transactional 加于接口方法后, 被本类普通接口方法通过接口调用, 有效
	 *                <p>
	 * @Transactional 加于接口方法后, 被它类的接口方法调用, 有效
	 *                <p>
	 * @Transactional 加于接口方法后, 被它类的私有方法调用后, 有效
	 * @return
	 */
	@Override
	@Transactional
	public boolean updateSeckillGoodsByLock(SeckillRequest seckillRequest, int count) {

		Seckill seckill = new Seckill();
		seckill.setNumber(count - 1);
		SeckillExample example = new SeckillExample();
		example.createCriteria().andSeckillIdEqualTo(seckillRequest.getSeckillId());

		return seckillMapper.updateByExample(seckill, example) > 0;
	}

	@Override
	public SeckillResponse seckillByCache(SeckillRequest seckillRequest) {
		return this.handleSeckill(seckillRequest);
	}

	private SeckillResponse handleSeckill(SeckillRequest seckillRequest) {
		String cacheCountKey = RedissonUtil.getRedisKey(NeptuneConstant.REDIS_SECKILL_GOODS_COUNT,
				seckillRequest.getSeckillId());
		SeckillUser user = WebUtil.getLoginUser();
		String cacheUserKey = RedissonUtil.getRedisKey(NeptuneConstant.REDIS_SECKILL_USER_ORDER,
				user.getId() + CharacterConstant.HYPHEN + seckillRequest.getSeckillId());
		if (hasSeckilledBefore(cacheUserKey, user.getId())){
			return new SeckillResponse(SeckillStatus.HAS_SECKILL.getMsg(),
					SeckillStatus.HAS_SECKILL.getStatus());
		}
		// 已售罄
		if (redisRepository.decrementReturnint(cacheCountKey,
				neptuneConfig.seckillInfoCacheTime) <= 0){
			return new SeckillResponse(SeckillStatus.SOLD_OUT.getMsg(),
					SeckillStatus.SOLD_OUT.getStatus());
		}

		String cacheInfoKey = RedissonUtil.getRedisKey(NeptuneConstant.REDIS_SECKILL_GOODS_INFO,
				seckillRequest.getSeckillId());
		Seckill seckill = redisRepository.get(cacheInfoKey, Seckill.class);
		// 活动未开始
		if (null != seckill.getStartTime()
				&& System.currentTimeMillis() < seckill.getStartTime().getTime()){
			return new SeckillResponse(SeckillStatus.NOT_START.getMsg(),
					SeckillStatus.NOT_START.getStatus());
		}
		// 活动已结束
		if (null != seckill.getEndTime()
				&& System.currentTimeMillis() > seckill.getEndTime().getTime()){
			return new SeckillResponse(SeckillStatus.END.getMsg(), SeckillStatus.END.getStatus());
		}

		CacheBean cacheBean = new CacheBean();
		cacheBean.setUserId(user.getId());
		cacheBean.setUserName(user.getUsername());
		cacheBean.setSeckillId(seckillRequest.getSeckillId());
		// redis缓存用户订单，防止重复秒杀
		redisRepository.set(cacheUserKey, cacheBean, neptuneConfig.seckillCacheTimePay);
		// 秒杀成功，发送信息到消息队列，更新数据库
		seckillProducer.sender(cacheBean);
		// 发送到死信队列 超时未付款 取消订单
		seckillDlxProducer.sender(cacheBean);
		// 直接返回秒杀信息
		return new SeckillResponse(SeckillStatus.SUCCESS.getMsg(),
				SeckillStatus.SUCCESS.getStatus());
	}

	/**
	 * 是否已经秒杀过,包含成功支付
	 * 
	 * @return
	 */
	private boolean hasSeckilledBefore(String key, String userId) {
		String cacheUserpAIDKey = RedissonUtil.getRedisKey(NeptuneConstant.REDIS_SECKILL_USER_PAID,
				userId);
		return redisRepository.hasKey(key) || redisRepository.hasKey(cacheUserpAIDKey);
	}

	private SeckillLog generateSeckillLog(SeckillRequest seckillRequest) {
		SeckillLog seckillLog = new SeckillLog();
		seckillLog.setUserId(seckillRequest.getUserId());
		seckillLog.setSeckillId(seckillRequest.getSeckillId());
		seckillLog.setCreateTime(new Date());
		seckillLog.setState(Byte.valueOf(String.valueOf(NeptuneConstant.STATUS_SUCCESS)));
		return seckillLog;
	}
}
